package com.frinder.official.akka.routing;

import akka.routing.Routee;
import akka.routing.RoutingLogic;
import com.frinder.official.utils.Utils;
import com.google.common.collect.Maps;
import lombok.extern.slf4j.Slf4j;
import scala.collection.Iterator;
import scala.collection.immutable.IndexedSeq;

import java.util.Map;

import static com.frinder.official.akka.routing.PayloadActor.Payload;

/**
 * Created by frinder on 2017/9/30.
 * <p>
 * simulation crc32 get remainder to split different request to dedicated routee
 */
@Slf4j
public class Crc32RoutingLogic implements RoutingLogic {

    private final Map<Long, Routee> routeeMap = Maps.newConcurrentMap();

    private final int roundSize;

    public Crc32RoutingLogic(int roundSize) {
        this.roundSize = roundSize;
    }

    @Override
    public Routee select(Object message, IndexedSeq<Routee> routees) {
        Routee routee = null;
        if (message instanceof Payload) {
            long remainder = Utils.remainder(((PayloadActor.Payload) message).getKey(), roundSize);
            if (null != (routee = routeeMap.get(remainder))) {
            } else {
                if (roundSize <= routees.length()) {
                    Iterator<Routee> iterable = routees.iterator();
                    Routee tee;
                    long i = 0;
                    while (iterable.hasNext()) {
                        tee = iterable.next();
                        routeeMap.put(i, tee);
                        i++;
                    }
                    routee = routeeMap.get(remainder);
                } else {
                    log.error("roundSize: {} greater than routees.length: {} !");
                }
            }
        } else {
            log.error("message type is: {}, need: {}", message.getClass(), Payload.class);
        }
        return routee;
    }
}
